package com.atguigu.day09;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import static org.apache.flink.table.api.Expressions.$;

public class Example4 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<Tuple2<String, String>> stream = env
                .fromElements(
                        Tuple2.of("Mary", "./home"),
                        Tuple2.of("Bob", "./cart"),
                        Tuple2.of("Mary", "./prod?id=1"),
                        Tuple2.of("Liz", "./home")
                );

        EnvironmentSettings settings = EnvironmentSettings.newInstance().inStreamingMode().build();
        StreamTableEnvironment streamTableEnvironment = StreamTableEnvironment.create(env, settings);
        Table table = streamTableEnvironment.fromDataStream(
                stream,
                $("f0").as("user"),
                $("f1").as("url")
        );
        streamTableEnvironment.createTemporaryView("clicks",table);
        //在临时视图上进行查询
        Table queryTable = streamTableEnvironment.sqlQuery("select user,count(url) from clicks group by user");
        //由于查询结果中存在更新操作、使用 toChangelogStream
        DataStream<Row> rowDataStream = streamTableEnvironment.toChangelogStream(queryTable);
        //-U 撤回数据
        //+U 更新数据
        rowDataStream.print();

        stream
                .map(x->Tuple2.of(x.f0,1L))
                .returns(Types.TUPLE(Types.STRING,Types.LONG))
                .keyBy(r->r.f0)
                .sum(1)
                .print();

        env.execute();
    }
}
